package com.zlsy.springbootdemo.rabbitmq.consumer;

import cn.hutool.core.date.DateUtil;
import com.zlsy.springbootdemo.rabbitmq.stream.Sink;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;

/**
 * @author zhouliang
 * @date 2020/12/10 14:47
 */
@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class XDeathDemoListener {

    /**
     * springCloud stream 获取重试次数
     */
    @StreamListener(Sink.X_DEATH_INPUT)
    public void listen(Message<String> in, @Header(name = "x-death", required = false) Map<?, ?> death) {
        log.warn("current time:{},msg：{},death:{}", DateUtil.now(), in, death);
        if (Objects.nonNull(death) && death.get("count").equals(3L)) {
            //丢弃消息 不重试了
            throw new ImmediateAcknowledgeAmqpException("Failed after 4 attempts");
        }
        //让消息重试
        throw new AmqpRejectAndDontRequeueException("failed");
    }

}
